PROTON-1442: [Cpp] Support for local transactions #437

DreamPearl · 2024-11-20T08:57:17Z

PROTON-1442

AMQP Transaction Sequence:

Declare transaction:

Client establishes link to Broker (Transaction resource) to target with transaction coordinator type (usual types are sender/receiver) (ATTACH frame)
Client (Transaction Controller) sends a special message to that link to create transaction (TRANSFER frame)
Broker returns a disposition with the transaction id (DISPOSITION frame)

Send Message in the transaction:

Client sends a message to the broker with transaction id in the state field.

Commit/Abort

* Added an extra handler to the python binding so that we can handle transactioned dispositions

* Modified the Python example broker so that it understands transaction requests, prints some useful output about what is happening, but doesn't honor the transaction semantics. It will queue up transactioned messages immediately and also doesn't correctly handle outgoing message releases (but it doesn't for non-transactioned messages either

* Make it compile * Make it fit the existing software structure better

astitcher

I've taken a fairly thorough look. If you have any questions about these review items we should discuss further.

astitcher · 2024-12-04T20:17:41Z

cpp/include/proton/container.hpp

@@ -312,6 +312,7 @@ class PN_CPP_CLASS_EXTERN container {
    /// Cancel task for the given work_handle.
    PN_CPP_EXTERN void cancel(work_handle);

+    PN_CPP_EXTERN transaction declare_transaction(proton::connection conn, proton::transaction_handler &handler, bool settle_before_discharge = false);


Since you take a connection as the first parameter here it probably makes more sense to move this entire method into the proton::connection class.
Actually maybe this should be a method on session to match the cms api, in either case you can find the connection easily - as a result it would make sense to hold the transaction as state in the connection or session and only allow a single transaction at that level.
In this case it will be easy enough to find the transaction as you can always navigate upwards from the delivery to find its session and on to the connection.

Moved the entire declare_transaction method to proton::session class.

astitcher · 2024-12-04T20:21:49Z

cpp/include/proton/target_options.hpp

+    PN_CPP_EXTERN target_options& type(const int);
+


This should not be needed. Introduce a new coordinator class that is peer to sender and receiver

cpp/include/proton/tracker.hpp

astitcher · 2024-12-04T20:24:43Z

cpp/include/proton/transaction.hpp

+class transaction_impl {
+  public:
+    proton::sender txn_ctrl;
+    proton::transaction_handler *handler = nullptr;
+    proton::binary id;
+    proton::tracker _declare;
+    proton::tracker _discharge;
+    bool failed = false;
+    std::vector<proton::tracker> pending;
+
+    void commit();
+    void abort();
+    void declare();
+    proton::tracker send(proton::sender s, proton::message msg);
+
+    void discharge(bool failed);
+    void release_pending();
+    void accept(delivery &d);
+    void update(tracker &d, uint64_t state);
+    void set_id(binary _id);
+
+    proton::tracker send_ctrl(proton::symbol descriptor, proton::value _value);
+    void handle_outcome(proton::tracker t);
+    transaction_impl(proton::sender &_txn_ctrl,
+                     proton::transaction_handler &_handler,
+                     bool _settle_before_discharge);
+
+    // delete copy and assignment operator to ensure no copy of this object is
+    // every made transaction_impl(const transaction_impl&) = delete;
+    // transaction_impl& operator=(const transaction_impl&) = delete;
+};


Create an internal header file for this class

Actually as I said above this entire class definition can probably go into transaction.cpp, there may not need to be a separate declaration.

astitcher · 2024-12-04T20:37:59Z

cpp/include/proton/transaction.hpp

+class
+PN_CPP_CLASS_EXTERN transaction_handler {
+  public:
+    PN_CPP_EXTERN virtual ~transaction_handler();
+
+    /// Called when a local transaction is declared.
+    PN_CPP_EXTERN virtual void on_transaction_declared(transaction);
+
+    /// Called when a local transaction is discharged successfully.
+    PN_CPP_EXTERN virtual void on_transaction_committed(transaction);
+
+    /// Called when a local transaction is discharged unsuccessfully (aborted).
+    PN_CPP_EXTERN virtual void on_transaction_aborted(transaction);
+
+    /// Called when a local transaction declare fails.
+    PN_CPP_EXTERN virtual void on_transaction_declare_failed(transaction);
+
+    /// Called when the commit of a local transaction fails.
+    PN_CPP_EXTERN virtual void on_transaction_commit_failed(transaction);
+};
+
+} // namespace proton
+


I think this should be in a different header file like message and messaging_handler

So if the API makes transactions hidden in a session then these callbacks either go away or instead become session callbacks: on_session_transaction_committed(session&), etc. I think transaction_declared() goes away entirely and it's purpose is now another use for on_session_open(session&). on_.._declare_failed should be handled by the on_session_error(session&).

astitcher · 2024-12-04T23:16:06Z

cpp/include/proton/transaction.hpp

+// TODO: This should not be accessible to users.
+class transaction_impl {
+  public:
+    proton::sender txn_ctrl;


I think this is one per connection not per transaction. The transaction needs a way to find its conection (or session if it's per session) anyway so you should always be able to find the transaction controller

astitcher · 2024-12-04T23:17:27Z

cpp/include/proton/transaction.hpp

+    proton::tracker _declare;
+    proton::tracker _discharge;


These must be exclusive as you cannot be delivering a declare and a discharge at the same time so I think you only need to keep track of a single outgoing delivery.

cpp/include/proton/transaction.hpp

cpp/examples/tx_recv.cpp

astitcher · 2024-12-04T23:30:29Z

cpp/examples/tx_recv.cpp

+    proton::transaction transaction;
+    proton::connection connection;


We shouldn't need to hold the connection as we should always be able to find the connection in a handler. This may be true for the transaction too - either we restrict the API to be one transaction per session or one transaction per connection. I think per session is porbably better as it matches the cms/jms api.

astitcher

I've not reviewed this comprehensively, but I think I'm now very much leaning towards not making the transaction class visible to the API user at all and having the transaction methods on the session. This is essentially the API in JMS and CMS.

astitcher · 2024-12-18T22:15:53Z

cpp/include/proton/transaction.hpp

+
+class transaction_handler;
+
+// TODO: This should not be accessible to users.


I think the entirety of transaction_impl should be in transaction.cpp.

astitcher · 2024-12-18T22:17:15Z

cpp/include/proton/transaction.hpp

+class transaction_impl {
+  public:
+    proton::sender txn_ctrl;
+    proton::transaction_handler *handler = nullptr;
+    proton::binary id;
+    proton::tracker _declare;
+    proton::tracker _discharge;
+    bool failed = false;
+    std::vector<proton::tracker> pending;
+
+    void commit();
+    void abort();
+    void declare();
+    proton::tracker send(proton::sender s, proton::message msg);
+
+    void discharge(bool failed);
+    void release_pending();
+    void accept(delivery &d);
+    void update(tracker &d, uint64_t state);
+    void set_id(binary _id);
+
+    proton::tracker send_ctrl(proton::symbol descriptor, proton::value _value);
+    void handle_outcome(proton::tracker t);
+    transaction_impl(proton::sender &_txn_ctrl,
+                     proton::transaction_handler &_handler,
+                     bool _settle_before_discharge);
+
+    // delete copy and assignment operator to ensure no copy of this object is
+    // every made transaction_impl(const transaction_impl&) = delete;
+    // transaction_impl& operator=(const transaction_impl&) = delete;
+};


Actually as I said above this entire class definition can probably go into transaction.cpp, there may not need to be a separate declaration.

astitcher · 2024-12-18T22:19:10Z

cpp/include/proton/transaction.hpp

+  public:
+    // TODO:
+    // PN_CPP_EXTERN transaction(transaction &o);
+    PN_CPP_EXTERN transaction();


Thinking it through I don't think that the user should be able to make a transaction at all, so take this out of the interface. I think the only way to get a transaction should be as the result of declaring a transaction.

Having said that I think a better overall API might be not to expose the transaction class to the API user at all and to have all the methods either on session or session::commit(), session::abort(), session::is_transactioned(). I think the way to do this without introducing a new transacted_session class would be to add a new session_option - something like bool transactioned() to make a transactioned session. Then in the transactioned case only call on_session_open when the transaction has been successfully declared. This would be instead of the transaction_handler::transaction_declared() callback.
Now any sender or receiver that is created from this session transparently will send or acknowledge in the transaction.

astitcher · 2024-12-18T23:10:52Z

cpp/include/proton/transaction.hpp

+class
+PN_CPP_CLASS_EXTERN transaction_handler {
+  public:
+    PN_CPP_EXTERN virtual ~transaction_handler();
+
+    /// Called when a local transaction is declared.
+    PN_CPP_EXTERN virtual void on_transaction_declared(transaction);
+
+    /// Called when a local transaction is discharged successfully.
+    PN_CPP_EXTERN virtual void on_transaction_committed(transaction);
+
+    /// Called when a local transaction is discharged unsuccessfully (aborted).
+    PN_CPP_EXTERN virtual void on_transaction_aborted(transaction);
+
+    /// Called when a local transaction declare fails.
+    PN_CPP_EXTERN virtual void on_transaction_declare_failed(transaction);
+
+    /// Called when the commit of a local transaction fails.
+    PN_CPP_EXTERN virtual void on_transaction_commit_failed(transaction);
+};
+
+} // namespace proton
+


So if the API makes transactions hidden in a session then these callbacks either go away or instead become session callbacks: on_session_transaction_committed(session&), etc. I think transaction_declared() goes away entirely and it's purpose is now another use for on_session_open(session&). on_.._declare_failed should be handled by the on_session_error(session&).

astitcher · 2024-12-18T23:15:10Z

cpp/include/proton/transfer.hpp

+    PN_CPP_EXTERN void transaction(transaction t);
+
+    PN_CPP_EXTERN class transaction transaction() const;


I think these are not use visible. They are only used internally in the implementation - especially if the transaction is hidden inside the session.

pematous · 2025-01-21T15:33:35Z

cpp/examples/tx_send.cpp

+        send(s);
+    }
+
+    void send(proton::sender &s) {


sender is defined as tx_recv class attribute, so I believe we don't need to pass the sender to send method.

pematous · 2025-01-21T15:35:45Z

cpp/examples/tx_recv.cpp

+        transaction.accept(d);
+        current_batch += 1;
+        if(current_batch == batch_size) {
+            transaction = proton::transaction(); // null


I believe we should do rather commit here, that way it works as expected and the receiver is closed after expected number of messages received (with the current implementation it's not).

pematous · 2025-01-21T15:37:19Z

cpp/examples/tx_recv.cpp

+
+int main(int argc, char **argv) {
+    std::string address("127.0.0.1:5672/examples");
+    int message_count = 9;


I would expect the defaults to be the same for tx_send and tx_recv (currently tx_send defaults to 6 messages while tx_recv to 9)

astitcher and others added 5 commits November 20, 2024 14:05

WIP: Python work for better transaction support

04eb16e

* Added an extra handler to the python binding so that we can handle transactioned dispositions

Implement local transactions

baa68ce

WIP: Fix up to transactions work:

ea79f47

* Make it compile * Make it fit the existing software structure better

Make example compile

ed624cd

DreamPearl marked this pull request as draft November 20, 2024 08:57

DreamPearl force-pushed the local-transactions branch from 9c4c6d3 to 74818fc Compare November 21, 2024 05:24

DreamPearl added 2 commits November 21, 2024 10:57

Add txn id in disposition frame bits

0e20334

Add transaction commit and abort functionalities

eb4514b

DreamPearl force-pushed the local-transactions branch from 74818fc to eb4514b Compare November 21, 2024 18:28

DreamPearl force-pushed the local-transactions branch from 3e96834 to fa9236d Compare December 3, 2024 12:56

astitcher requested changes Dec 4, 2024

View reviewed changes

Add tx_recv file

347cf6e

DreamPearl force-pushed the local-transactions branch from fa9236d to 347cf6e Compare December 16, 2024 13:29

astitcher requested changes Dec 18, 2024

View reviewed changes

pematous reviewed Jan 21, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PROTON-1442: [Cpp] Support for local transactions #437

PROTON-1442: [Cpp] Support for local transactions #437

DreamPearl commented Nov 20, 2024 •

edited

Loading

astitcher left a comment

astitcher Dec 4, 2024

DreamPearl Dec 16, 2024

astitcher Dec 4, 2024

astitcher Dec 4, 2024

astitcher Dec 18, 2024

astitcher Dec 4, 2024

astitcher Dec 18, 2024

astitcher Dec 4, 2024

astitcher Dec 4, 2024

astitcher Dec 4, 2024

astitcher left a comment

astitcher Dec 18, 2024

astitcher Dec 18, 2024

astitcher Dec 18, 2024

astitcher Dec 18, 2024

astitcher Dec 18, 2024

astitcher Dec 18, 2024

pematous Jan 21, 2025

pematous Jan 21, 2025

pematous Jan 21, 2025

		proton::transaction transaction;
		proton::connection connection;


		class transaction_handler;

		// TODO: This should not be accessible to users.

		PN_CPP_EXTERN void transaction(transaction t);

		PN_CPP_EXTERN class transaction transaction() const;

PROTON-1442: [Cpp] Support for local transactions #437

Are you sure you want to change the base?

PROTON-1442: [Cpp] Support for local transactions #437

Conversation

DreamPearl commented Nov 20, 2024 • edited Loading

astitcher left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

astitcher left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

DreamPearl commented Nov 20, 2024 •

edited

Loading